package com.actor;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;

/**
 * @author sad
 */
public abstract class Actor<T> implements ActorRef<T> {
    private int futureCounter = 1;
    private final ActorSystem actorSystem;
    private QueueProcessor queue;
    private final String actorName;
    private final Map<Integer, Future> futures = new HashMap<>();

    public Actor(String name, ActorSystem actorSystem) {
        this.actorSystem = actorSystem;
        this.actorName = name;
        init();
    }

    public Map<Integer, Future> getFutures() {
        return futures;
    }

    private void init() {
        actorSystem.registerActor(this);
    }

    public ActorSystem getActorSystem() {
        return actorSystem;
    }

    public String getActorName() {
        return actorName;
    }

    public void setActorQueue(QueueProcessor actorQueue) {
        this.queue = actorQueue;
    }

    public QueueProcessor getActorQueue() {
        return queue;
    }

    public void stop() {
        actorSystem.unregisterActor(this);
    }

    @Override
    public void tell(Object object, ActorRef sender) {
        MessageWrapper mw = new MessageWrapper(object, sender);
        mw.setReceiver(this);
        queue.addMessage(mw);
    }

    @Override
    public void tell(MessageWrapper messageWrapper) {
        messageWrapper.setReceiver(this);
        queue.addMessage(messageWrapper);
    }

    public Future ask(ActorRef receiver, Object value, ActorRef sender) {
        return ask(receiver, value, sender, -1);
    }

    public Future ask(ActorRef receiver, Object value, ActorRef sender, long timeout) {
        int futureTag = futureCounter;
        futureCounter++;
        if (futureCounter > 2000000000) {
            futureCounter = 1;
        }

        FutureActorRef futureActorRef = new FutureActorRef(this, futureTag);
        MessageWrapper mw = new MessageWrapper(value, futureActorRef);
        receiver.tell(mw);
        Future future = new Future();
        futures.put(futureTag, future);
        if (timeout != -1) {
            scheduleFutureTimeout(futureActorRef, timeout);
        }
        return future;
    }

    private void scheduleFutureTimeout(final FutureActorRef futureActorRef, long timeout) {
        ScheduledFuture scheduledFuture = getActorSystem().getScheduler().schedule(new Runnable() {
            @Override
            public void run() {
                futureActorRef.onError(new FutureTimeoutException("Future timeoutexception"), null, getActorSystem().getNullActor());
            }
        }, timeout, TimeUnit.MILLISECONDS);
        futureActorRef.setTimeoutFuture(scheduledFuture);
    }

    public abstract void receive(T value, ActorRef sender);

    @Override
    public void onError(Throwable throwable, T value, ActorRef sender) {
        throwable.printStackTrace();
    }

    @Override
    public String toString() {
        return "Actor " + getActorName();
    }

}
